package epoch.game.core.distr;

import java.io.IOException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import epoch.game.core.buffer.IBuffer;
import epoch.game.core.distr.config.DConfig;
import epoch.game.core.message.IMessage;
import epoch.game.core.message.SysMessageIds;
import epoch.game.core.serialize.StreamReader;
import epoch.game.core.server.ISession;

/**
 * 服务节点消息通道
 * @author will
 * @date 2015年11月25日 下午3:20:00
 */
public class NodeChannel implements INodeChannel {
	
	private static Logger logger = LogManager.getLogger(NodeChannel.class);
	
	//session中缓存的nodeId
	public static final String node_id="node_id";
	//节点
	private Node node;

	private ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(1);
	
	public NodeChannel(Node node) {
		this.node = node;
        int intv = 20;
		scheduledExecutor.scheduleAtFixedRate(()->{
			if(!node.isAvailable()){
				return;
			}
            for (RemoteNode remote: node.getRemoteNodes().values()) {
            	if (remote.isConnected()) {
            		remote.flushSendMsg();
            	}
            }
		}, intv, intv, TimeUnit.MILLISECONDS);
	}
	
	
	public Node getNode() {
		return node;
	}

	public void setNode(Node node) {
		this.node = node;
	}

	private void handleReg(ISession session, IBuffer buffer) {
		NodeRegisterMessage msg = new NodeRegisterMessage();
		StreamReader reader = new StreamReader(buffer);
		try {
			msg.readFrom(reader);
		} catch (IOException e) {
		}

		if(!DConfig.NODE_SECRET_KEY.equals(msg.getSecretKey())){
			logger.error("Node register failed: secretKey error: {}", msg.getSecretKey());
			session.close();
			return;
		}
		int fromNodeId = msg.getNodeId();
		RemoteNode remoteNode = node.getRemoteNode(fromNodeId);
		if (remoteNode == null) {
			remoteNode = new RemoteNode(msg.getHost(), msg.getPort());
			remoteNode.setId(fromNodeId);
			remoteNode.setName(msg.getNodeName());
			remoteNode.setType(msg.getType());
			node.addRemoteNode(remoteNode);
		} else {
			logger.info("Node registered override old: {}", remoteNode.getTag());
			remoteNode.setHost(msg.getHost());
			remoteNode.setPort(msg.getPort());
			remoteNode.setName(msg.getNodeName());
			remoteNode.setType(msg.getType());
			remoteNode.setNeedRetry(node.needReconnect(remoteNode));
		}
		remoteNode.setSession(session);
		remoteNode.setRegistered(true);
		session.setAttribute(node_id, fromNodeId);
	}
	
	@Override
	public void onMessageReceived(ISession session, byte[] data) {
		try {
			IBuffer buffer = IBuffer.wrap(data);
			short msgId = buffer.readShort();
			if (msgId == SysMessageIds.node_register) {
				handleReg(session, buffer);
			} else {
				if (msgId < 0) {
					NodeChannelUtil.handleRpc(node, session, msgId, buffer);
				} else {
					Integer fromNodeId = session.getAttribute(node_id);
					if (fromNodeId != null) {
						RemoteNode remoteNode = node.getRemoteNode(fromNodeId);
						if (remoteNode != null) {
							NodeType fromNodeType = remoteNode.getType();
							IMessage iMessage = node.getHandler().newMessage(msgId, buffer);
							if (iMessage != null) {
								node.getHandler().onNodeMessageReveived(iMessage, fromNodeType);
							}
						}
					}
				}
			}
		} catch (Exception ex) {
			logger.error("处理收到的消息出错！"+ex, ex);
		}
	}
	
	@Override
	public void onSessionClosed(ISession session) {		
		logger.info("Node session closed: {}", session);
		if (session.containsAttribute(node_id) &&
			session.getAttribute(node_id) != null) {
			int nodeId = session.getAttribute(node_id);
			RemoteNode remoteNode = node.getRemoteNode(nodeId);
			if (remoteNode != null) {
				remoteNode.delSession(session);
			}
		}
		node.getHandler().onNodeSessionClosed(session);
	}

	@Override
	public void onSessionCreated(ISession session) {
		logger.info("Node session created: {}", session);
		node.getHandler().onNodeSessionCreated(session);
	}

	@Override
	public void onSessionOpened(ISession session) {
		logger.info("Node send register msg: {}", session);
		NodeChannelUtil.sendRegMsg(node, session);
	}

	@Override
	public void onSessionIdle(ISession session) {
		node.getHandler().onNodeSessionIdle(session);
	}

	@Override
	public void onExceptionCaught(ISession session, Throwable throwable) {
		node.getHandler().onNodeExceptionCaught(session,throwable);
	}

	@Override
	public void onMessageSent(ISession session, Object message) {
		node.getHandler().onNodeMessageSent(session, message);
	}


}
